package cn.me.alphamq.nameserver.task.cluster.chain;

import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.nameserver.cache.Container;
import cn.me.alphamq.nameserver.event.model.cluster.ReplicationMsgEvent;
import cn.me.alphamq.nameserver.store.ReplicationMsgQueueManager;
import cn.me.alphamq.nameserver.task.cluster.ReplicationTask;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 链式复制中非尾节点给下一个节点发送复制消息的任务类
 *
 * @author f
 */
public class NodeSendReplicationMsgTask extends ReplicationTask {

    private static final Logger log = LoggerFactory.getLogger(NodeSendReplicationMsgTask.class);

    @Override
    protected void replicationTask() {
        log.info("start sending replication msg task...");
        ReplicationMsgQueueManager replicationMsgQueueManager = Container.getReplicationMsgQueueManager();

        while (true) {
            try {
                ReplicationMsgEvent replicationMsgEvent = replicationMsgQueueManager.get();

                // 确保下一个节点是可用的
                Channel nextNodeChannel = Container.getMultiFunctionalChannel();
                if (nextNodeChannel.isActive()) {
                    // 给下一个节点发送复制消息
                    nextNodeChannel.writeAndFlush(new TcpMsg(EventCodeEnum.NODE_REPLICATION_MSG_RESP,
                            JSON.toJSONBytes(replicationMsgEvent)));
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
}
